import akka.actor.Actor
import message._
import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
import org.apache.curator.framework.api.{BackgroundCallback, CuratorEvent, CuratorEventType}
import org.apache.curator.framework.recipes.leader.{LeaderSelector, LeaderSelectorListenerAdapter}
import org.apache.curator.framework.state.{ConnectionState, ConnectionStateListener}
import org.apache.curator.retry.RetryNTimes
import org.apache.zookeeper.CreateMode
import org.apache.zookeeper.data.Stat

class ZkClient extends Actor{

    val ZK_PATH = "/zkTest"
    val CLIENT_PATH = ZK_PATH + "/client"
    val MASTER_PATH = "/clientMaster"
    val ZK_ADDRESS = "127.0.0.1:2181"
    var clientName: String = null
    val zkClient = CuratorFrameworkFactory.newClient(ZK_ADDRESS, new RetryNTimes(10, 500))
    val data = "data1"
    zkClient.start()

    val selector = new LeaderSelector(zkClient, MASTER_PATH, new LeaderSelectorListenerAdapter {
        override def takeLeadership(client: CuratorFramework) = {
            println("client: ", clientName, " become master")
            Thread.sleep(3000)
            println("release master")
        }
    })
    selector.autoRequeue()

    /***
      * 建立一个ZKClient客户端
      * 并且建立一个临时节点
      * 临时节点创建后不能在server中马上看到
      * 必须监视节点的存在
      * 节点创建成功后 申请master
      */
    zkClient.create().
        creatingParentsIfNeeded().
        withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
        inBackground(new BackgroundCallback {
            override def processResult(client: CuratorFramework, event: CuratorEvent) = {
                event.getType match {
                    case CuratorEventType.CREATE =>
                        clientName = event.getName
                        println("create success clientName: ", clientName)
                        selector.start()
                        Server.countDownLatch.countDown()
                    case _ =>
                }
            }
        }).
        forPath(CLIENT_PATH, data.getBytes())

    //zk断线重连  重新生成节点
    zkClient.getConnectionStateListenable.addListener(new ConnectionStateListener {
        override def stateChanged(client: CuratorFramework, newState: ConnectionState) = {
            if(newState == ConnectionState.RECONNECTED) {
                zkClient.create().
                    creatingParentsIfNeeded().
                    withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                    forPath(CLIENT_PATH, data.getBytes())
            }
        }
    })

    override def receive: Receive = {
        case CloseClient => zkClient.close()
        case ShowNode =>
            println(zkClient.getData.forPath(CLIENT_PATH))
        case NodeList =>
            println(zkClient.getChildren.forPath(ZK_PATH))
        case DeleteNode =>
            zkClient.delete().forPath(CLIENT_PATH)
        case UpdateData =>
            zkClient.setData().forPath(CLIENT_PATH)
        case _ =>
    }

}